// 网络通信
#pragma once
#include "detail.hpp"
#include "field.hpp"
#include "abstruct.hpp"
#include "message.hpp"
#include <muduo/net/TcpServer.h>
#include <muduo/net/TcpClient.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/base/CountDownLatch.h>
#include <mutex>
#include <unordered_map>

namespace bitRPC
{
    class MuduoBuffer : public BaseBuffer
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;
        MuduoBuffer(muduo::net::Buffer *buffer) : _buffer(buffer) {}
        virtual size_t readableSize() override // 缓冲区的长度
        {
            return _buffer->readableBytes();
        }
        virtual int32_t peekInt32() override // 读取四字节，不删除
        {
            // muduo 库是一哥网络库，会将我们从缓冲区中取出来的四字节的字节序数据
            //  转换为主机字节序
            return _buffer->peekInt32();
        }
        virtual void retrieveInt32() override // 移除前四节的数据
        {
            _buffer->retrieveInt32();
        }
        virtual int32_t readInt32() override // 读取四字节并删除
        {
            return _buffer->readInt32();
        }
        virtual std::string retrieveAsString(size_t len) override
        {
            return _buffer->retrieveAsString(len);
        }

    private:
        muduo::net::Buffer *_buffer;
    };
    class BufferFactory
    {
    public:
        template <typename... Args>
        static BaseBuffer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    class LVProtocol : public BaseProtocol
    {
    public:
        // 接收到的消息
        using ptr = std::shared_ptr<LVProtocol>;
        //|--Len--|Value|
        //|--Len--|mytype|lenid|id|body|
        virtual bool onMessage(BaseBuffer::ptr &buffer, BaseMessage::ptr &msg) override
        {
            // 默认消息是能够处理的
            int32_t total = buffer->peekInt32();
            MType mytype = (MType)buffer->readInt32();
            int32_t lenid = buffer->readInt32();
            std::string id = buffer->retrieveAsString(lenid);
            int32_t body_len = total - lenfiledLenth - idlenfiledLenth - mytypefiledLenth;
            std::string body = buffer->retrieveAsString(body_len);
            msg = MessageFactory::create(mytype);
            if (msg == nullptr)
            {
                ELOG("消息类型错误，构造失败");
                return false;
            }
            msg->setId(id);
            bool ret = msg->unserialize(body);
            if (ret == false)
            {
                ELOG("消息反序列化失败");
                return false;
            }
            return true;
        }
        virtual bool canProcess(BaseBuffer::ptr &buffer) override // 当前缓冲区的能否处理
        {
            if (buffer->readableSize() < lenfiledLenth)
            {
                return false;
            }
            int32_t total = buffer->peekInt32();
            if (total + lenfiledLenth > buffer->readableSize())
            {
                return false;
            }
            return true;
        }
        //    发送的消息
        virtual std::string serialize(const BaseMessage::ptr &msg) override
        {
            std::string body = msg->serialize();
            std::string id = msg->Rid();
            auto mytype = htonl((int32_t)msg->mtype());
            int32_t idlen = htonl(id.size());
            int32_t htotal = mytypefiledLenth + idlenfiledLenth + id.size() + body.size();
            int32_t total = htonl(htotal);
            std::string result;
            result.reserve(total);
            result.append((char *)&total, lenfiledLenth);
            result.append((char *)&mytype, mytypefiledLenth);
            result.append((char *)&idlen, idlenfiledLenth);
            result.append(id);
            result.append(body);
            return result;
        }

    private:
        size_t lenfiledLenth = 4;
        size_t mytypefiledLenth = 4;
        size_t idlenfiledLenth = 4;
    };
    class ProtocolFactory
    {
    public:
        template <typename... Args>
        static std::shared_ptr<LVProtocol> create(Args &&...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };
    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;
        MuduoConnection(const muduo::net::TcpConnectionPtr &conn, const LVProtocol::ptr protocol) : _conn(conn),
                                                                                                    _protocol(protocol)
        {
        }
        virtual bool send(const BaseMessage::ptr &msg) override // 发送消息
        {
            std::string body = _protocol->serialize(msg);
            _conn->send(body);
            return true;
        }
        virtual bool shutdown() // 关闭连接
        {
            _conn->shutdown();
            return true;
            ;
        }
        virtual bool connected() override // 是否连接
        {
            return _conn->connected();
        }

    private:
        muduo::net::TcpConnectionPtr _conn;
        BaseProtocol::ptr _protocol;
    };
    class ConnectionFactory
    {
    public:
        template <typename... Args>
        static std::shared_ptr<BaseConnection> create(Args &&...args)
        {
            return std::make_shared<muduo::net::TcpConnection>(std::forward<Args>(args)...);
        }
    };
    class MuduoServer : public BaseServer
    {
    public:
        using ptr = std::shared_ptr<MuduoServer>;
        MuduoServer(int port) : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "MuduoServer",
                                        muduo::net::TcpServer::Option::kReusePort),
                                _protocol(ProtocolFactory::create())
        {
        }
        void start()
        {
            _server.setMessageCallback(std::bind(&MuduoServer::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
            _server.start();
            _baseloop.loop();
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &con)
        {
            if (con.get())
            {
                std::cout << "连接成功" << std::endl;
                auto mudo_con = ConnectionFactory::create(con, _protocol);
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    _conns[con] = mudo_con;
                }
                if (_cb_connection)
                    _cb_connection(mudo_con);
            }
            else
            {
                std::cout << "连接断开" << std::endl;
                BaseConnection::ptr muduo_con;
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    auto it = _conns.find(con);
                    if (it == _conns.end())
                    {
                        return;
                    }
                    muduo_con = it->second;
                    _conns.erase(it);
                }
                if (_cb_close)
                    _cb_close(muduo_con);
            }
        }
        void onMessage(const muduo::net::TcpConnectionPtr &conn,
                       muduo::net::Buffer *buf,
                       muduo::Timestamp t)
        {
            DLOG("有数据到来");
            BaseBuffer::ptr base_buf = BufferFactory::create(buf);
            while (1)
            {
                if (_protocol->canProcess(base_buf) == false)
                {
                    // 数据不足
                    if (base_buf->readableSize() > maxDatasize)
                    {
                        ELOG("缓冲区数据量过大");
                        conn->shutdown();
                        return;
                    }
                    break;
                }
                BaseMessage::ptr _msg;
                bool ret = _protocol->onMessage(base_buf, _msg);
                if (ret == false)
                {
                    ELOG("缓冲区的数据出错");
                    conn->shutdown();
                    return;
                }
                BaseConnection::ptr base_conn;
                {
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        conn->shutdown();
                        return;
                    }
                    base_conn = it->second;
                }
                if (_cb_message)
                    _cb_message(base_conn, _msg);
            }
        }
        size_t maxDatasize = (1 << 16);
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;
        BaseProtocol::ptr _protocol;
        std::mutex _mutex;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
    };
    class ServerFactory
    {
    public:
        template <typename... Args>
        static BaseServer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };
    class MuduoClient : public BaseClient
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;
        MuduoClient(const std::string &ip, int port) : _downlatch(1),
                                                       _baseloop(_loopthread.startLoop()),
                                                       _client(_baseloop, muduo::net::InetAddress(ip, port), "MuduoClient")
        {
            // 设置连接事件（连接建立/管理）的回调
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));
            // 设置连接消息的回调
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this,
                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _client.connect();
            _downlatch.wait();
        }
        virtual bool connect() override // 连接服务器
        {
            DLOG("设置回调函数，连接服务器");
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));
            // 设置连接消息的回调
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this,
                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            // 连接服务器
            _client.connect();
            _downlatch.wait();
            DLOG("连接服务器成功！");
            return true;
        }
        virtual bool shutdown() override
        {
            return _conn->shutdown();
        }
        virtual bool send(const BaseMessage::ptr &msg) override
        {
            if (connected() == false)
            {
                ELOG("连接已断开！");
                return false;
            }
            _conn->send(msg);
        }
        virtual BaseConnection::ptr connection() override
        {
            return _conn;
        }
        virtual bool connected()
        { // 是否连接
            return (_conn && _conn->connected());
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &con)
        {
            if (con->connected())
            {
                std::cout << "连接成功" << std::endl;
                // 计数减1
                _downlatch.countDown();
                _conn = ConnectionFactory::create(con, _protocol);
                if (_cb_connection)
                {
                    _cb_connection(_conn);
                }
            }
            else
            {
                std::cout << "连接断开" << std::endl;
                _conn.reset();
            }
        }

        void onMessage(const muduo::net::TcpConnectionPtr &conn,
                       muduo::net::Buffer *buf,
                       muduo::Timestamp t)
        {
            BaseBuffer::ptr base_buf = BufferFactory::create(buf);
            while (1)
            {
                if (_protocol->canProcess(base_buf) == false)
                {
                    // 数据不足
                    if (base_buf->readableSize() > maxDataSize)
                    {
                        ELOG("缓冲区数据量过大");
                        conn->shutdown();
                        return;
                    }
                    break;
                }
                BaseMessage::ptr _msg;
                bool ret = _protocol->onMessage(base_buf, _msg);
                if (ret == false)
                {
                    ELOG("缓冲区的数据出错");
                    conn->shutdown();
                    return;
                }
                // BaseConnection::ptr base_conn;
                if (_cb_message)
                    _cb_message(_conn, _msg);
            }
        }

    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        BaseConnection::ptr _conn;
        muduo::CountDownLatch _downlatch;
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop *_baseloop;
        muduo::net::TcpClient _client;
    };
    class ClientFactory
    {
    public:
        template <typename... Args>
        static BaseClient::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };
}